package com.sunday.common.mq.rabbit.study.base.e16_ConsumerPriorities;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.sunday.common.mq.rabbit.study.base.ConnectionUtil;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.junit.jupiter.api.Test;

import java.nio.charset.StandardCharsets;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.TimeUnit;

/**
 * https://www.rabbitmq.com/consumer-priority.html
 * 消费者优先次序
 * 消费者优先级允许您确保高优先级 使用者在活动时接收消息，其中包含消息 仅在高优先级时向优先级较低的消费者 消费者阻止。
 * 通常，连接到队列的活动使用者接收消息 以循环的方式从它开始。当消费者的优先事项是 使用中，如果多个活动消息处于轮循机制 消费者以相同的高优先级存在。
 *
 * 活跃消费者的定义
 * 上述段落将消费者称为活跃或被阻止。在任何时候，任何给定的消费者要么是 非此即彼。活跃的消费者是可以接收 无需等待的消息。如果消费者被阻止，则 无法接收消息 - 因为它的通道已到达 发出 Basic.QoS 后或仅仅是因为网络拥塞而未确认消息的最大数量。
 * 因此，对于每个队列，必须至少满足以下三项条件之一：
 *  1. 没有活跃的消费者
 *  2. 队列为空
 *  3. 队列正忙于向使用者传递消息
 * 请注意，使用者可以在活动和阻止许多之间切换 每秒次数。因此，我们不会暴露消费者是否 通过管理插件或 RabbitMQCTL 处于活动状态或被阻止。
 * 当使用消费者优先级时，您可以期望您的最高优先级 优先使用者接收所有消息，直到它们成为 阻止，此时优先级较低的消费者将开始 收到一些。重要的是要了解 RabbitMQ 会 仍然优先考虑传递消息 - 它不会等待 高优先级阻止的使用者，如果存在，则变为解锁 一个活跃的低优先级消费者就绪。
 *
 * 使用消费者优先级
 * 将 basic.consumption 方法中的 x-priority 参数设置为整数值。
 * 消费者 未指定值的优先级为 0。更大的数字 表示优先级更高，正数和负数 可以使用。
 */
@Slf4j
public class ConsumerPriorities {

    /**
     * https://www.rabbitmq.com/consumer-priority.html
     * 消费者优先次序
     * 消费者优先级允许您确保高优先级 使用者在活动时接收消息，其中包含消息 仅在高优先级时向优先级较低的消费者 消费者阻止。
     * 通常，连接到队列的活动使用者接收消息 以循环的方式从它开始。当消费者的优先事项是 使用中，如果多个活动消息处于轮循机制 消费者以相同的高优先级存在。
     */
    @Test
    @SneakyThrows
    public void consumerPriority() {

        String queue = UUID.randomUUID().toString();

        ConnectionFactory cf = ConnectionUtil.factory();
        try (Connection connection = cf.newConnection()) {
            Channel channel = connection.createChannel();
            /**
             * 优先级和资源使用情况的最大数量 x-max-priority
             * 如果需要优先级队列，我们建议使用 1 到 10 之间的队列。 当前使用更多优先级将通过使用更多的 Erlang 进程来消耗更多的 CPU 资源。运行时调度也会受到影响。
             */
            Map<String, Object> args = new HashMap<>();
            args.put("x-max-priority", 5);

            channel.queueDeclare(queue, false, false, false, args);

            for (int i = 0; i <= 10; i++) {
                AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
                        .priority(i)        //优先级
                        .build();
                channel.basicPublish("", queue, props, String.valueOf(i).getBytes(StandardCharsets.UTF_8));
            }

            /**
             * 使用消费者优先级
             * 将 basic.consumption 方法中的 x-priority 参数设置为整数值。
             * 消费者 未指定值的优先级为 0。更大的数字 表示优先级更高，正数和负数 可以使用。
             */
            channel.basicConsume(queue, true, Collections.singletonMap("x-priority", 2),
                    (consumerTag, delivery) -> {
                        String message = new String(delivery.getBody(), StandardCharsets.UTF_8);
                        log.info(" [level 2][x] Received '{}'", message);
                    },
                    consumerTag -> {
                    });

            channel.basicConsume(queue, true, Collections.singletonMap("x-priority", 1),
                    (consumerTag, delivery) -> {
                        String message = new String(delivery.getBody(), StandardCharsets.UTF_8);
                        log.info(" [level 1][x] Received '{}'", message);
                    },
                    consumerTag -> {
                    });

            /**
             * 本轮测试可知 :
             * 如果一个消费者的优先级高的时候，将会抢夺低优先级消费者的消息，
             * 并且只有高消费者阻塞那么才会把消息发送给低优先级的消费者。如果出现多个相同高优先级的消费者，
             * 那么消息只会在高优先级消费者之间轮询
             */
            TimeUnit.SECONDS.sleep(2);

        }

    }

}
